package com.lujianfei.rxjavademo;

import android.content.Context;
import android.util.Log;

import androidx.test.InstrumentationRegistry;
import androidx.test.runner.AndroidJUnit4;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.schedulers.Schedulers;

import org.junit.Test;
import org.junit.runner.RunWith;

import static org.junit.Assert.*;

/**
 * Instrumented test, which will execute on an Android device.
 *
 * @see <a href="http://d.android.com/tools/testing">Testing documentation</a>
 */
@RunWith(AndroidJUnit4.class)
public class ExampleInstrumentedTest {
    private static final String TAG = "ExampleInstrumentedTest";
    Object locker = new Object();

    @Test
    public void useAppContext() {
        // Context of the app under test.
        Context appContext = InstrumentationRegistry.getTargetContext();

        assertEquals("com.lujianfei.rxjavademo", appContext.getPackageName());
    }

    @Test
    public void testAsync() throws InterruptedException {
        synchronized (locker) {
            //创建被观察者
            Observable<String> stringObservable = Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    System.out.println("被观察者 " + Thread.currentThread().getName() +"=========================");
                    Thread.sleep(3000);
                    emitter.onNext("Response");
                }
            });
            //观察者
            Observer<String> observer = new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onNext(String s) {
                    System.out.println("观察者 " + s + " " + Thread.currentThread().getName() +"=========================");
                    locker.notifyAll();
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {
                }
            };
            stringObservable.subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(observer); //被观察者子线程  观察者主线程

            locker.wait();
        }
    }

    @Test
    public void testMultiAsync() throws InterruptedException {
        System.out.println("testMultiAsync=======================");
        synchronized (locker) {
            //被观察者1
            Observable<String> stringObservable = Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    System.out.println("被观察者1 " + Thread.currentThread().getName()+"========================");
                    Thread.sleep(1000);
                    emitter.onNext("StringParam");
                }
            });
            //被观察者2
            Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    System.out.println("被观察者2 " + Thread.currentThread().getName()+"========================");
                    Thread.sleep(1000);
                    emitter.onNext(1);
                }
            });
            //观察者
            Observer<Object> objectObserver = new Observer<Object>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onNext(Object o) {
                    System.out.println("观察者 " + Thread.currentThread().getName()+"========================");
                    locker.notifyAll();
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            };

            Observable.zip(stringObservable, integerObservable, new BiFunction<String, Integer, Object>() {
                @Override
                public Object apply(String s, Integer integer) throws Exception {
                    //被观察者3
                    System.out.println("被观察者3 "+Thread.currentThread().getName()+"========================");
                    return s + integer;
                }
            }).subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(objectObserver);
            locker.wait();
        }
    }
}
